package org.fastsyncer.manager.canel;

import java.net.InetSocketAddress;
import java.util.List;
import java.util.Map;

import org.json.JSONArray;
import org.json.JSONException;
import org.json.JSONObject;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.client.CanalConnectors;
import com.alibaba.otter.canal.protocol.CanalEntry.Column;
import com.alibaba.otter.canal.protocol.CanalEntry.Entry;
import com.alibaba.otter.canal.protocol.CanalEntry.EntryType;
import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.alibaba.otter.canal.protocol.CanalEntry.RowChange;
import com.alibaba.otter.canal.protocol.CanalEntry.RowData;
import com.alibaba.otter.canal.protocol.Message;

public class CanelRemoteClient {

	private static CanelRemoteClient instance = new CanelRemoteClient();

	private CanelRemoteClient() {
	}

	public static CanelRemoteClient getInstance() {
		return instance;
	}

	public JSONArray pull(Map<String, String> params) {
		String instance = params.get("instance");
		String database = params.get("database");
		String table = params.get("table");
		String ip = params.get("ip");
		int port = Integer.parseInt(params.get("port"));
		InetSocketAddress addr = new InetSocketAddress(ip, port);
		CanalConnector connector = CanalConnectors.newSingleConnector(addr, instance, "", "");

		JSONArray msg = null;
		try {
			connector.connect();
			/**
			 * 常见例子：
			 * 1.  所有表：.*   or  .*\\..*
			 * 2.  canal schema下所有表： canal\\..*
			 * 3.  canal下的以canal打头的表：canal\\.canal.*
			 * 4.  canal schema下的一张表：canal.test1
			 * 5.  多个规则组合使用：canal\\..*,mysql.test1,mysql.test2 (逗号分隔)
			 */
			//监听指定库下面的一张表
			connector.subscribe(database+"."+table);
			connector.rollback();

			Message message = connector.getWithoutAck(1000); // 获取指定数量的数据
			List<Entry> entries = message.getEntries();
			if (message.getId() != -1 && entries != null && entries.size() > 0) {
				msg = parseEntry(entries);
			}
			connector.ack(message.getId()); // 提交确认
			// connector.rollback(batchId); // 处理失败, 回滚数据
		} catch (Exception e) {
			e.printStackTrace();
		} finally {
			connector.disconnect();
		}
		return msg;
	}

	private static JSONArray parseEntry(List<Entry> entrys) {
		JSONArray msg = new JSONArray();
		for (Entry entry : entrys) {
			// 排除事务解析
			EntryType entryType = entry.getEntryType();
			if (entryType == EntryType.TRANSACTIONBEGIN || entryType == EntryType.TRANSACTIONEND) {
				continue;
			}

			try {
				// 获取表名
				//String table = entry.getHeader().getTableName();
				// 解析参数
				RowChange rowChage = RowChange.parseFrom(entry.getStoreValue());
				// 获取操作事件
				EventType eventType = rowChage.getEventType();

				// 排除非增删改操作
				if (eventType != EventType.DELETE && eventType != EventType.INSERT && eventType != EventType.UPDATE) {
					continue;
				}
				
				JSONObject row = new JSONObject();
				JSONArray beforeArr = new JSONArray();
				JSONArray afterArr = new JSONArray();
				// 获取执行参数
				List<RowData> rowDatasList = rowChage.getRowDatasList();
				// 获取执行前后参数
				for (RowData rowData : rowDatasList) {
					if (eventType == EventType.DELETE) {
						beforeArr = printColumn(rowData.getBeforeColumnsList());
					} else if (eventType == EventType.INSERT) {
						afterArr = printColumn(rowData.getAfterColumnsList());
					} else if (eventType == EventType.UPDATE) {
						beforeArr = printColumn(rowData.getBeforeColumnsList());
						afterArr = printColumn(rowData.getAfterColumnsList());
					}
				}
				row.put("eventType", eventType);
				row.put("before", beforeArr);
				row.put("after", afterArr);
				msg.put(msg.length(), row);
			} catch (Exception e) {
				throw new RuntimeException("解析binlog时异常:" + entry.toString(), e);
			}
		}
		return msg;
	}

	private static JSONArray printColumn(List<Column> columns) throws JSONException {
		JSONArray row = new JSONArray();
		for (Column col : columns) {
			JSONObject attr = new JSONObject();
			attr.put("name", col.getName());
			attr.put("value", col.getValue());
			row.put(row.length(), attr);
		}
		return row;
	}
}